package com.yujiahao.bigdata.test

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark01_TestHotTop10 {
  def main(args: Array[String]): Unit = {
    //TODO 1、获得Spark环境的连接

    val conf: SparkConf = new SparkConf().setAppName("TestTop10").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)

    //TODO 3、业务逻辑
    //需求开发的基本思路：1、缺什么补什么。2、多什么删什么

    //TODO 3.1) 获取原始数据
    val rdd: RDD[String] = sc.textFile("data/user_visit_action.txt")
    //TODO 3.2) 将原始数据统计聚合.
    // 1、统计品类的点击数量。clickCountData
    val clickCountData: RDD[(String, Int)] = rdd.filter(data => {
      val datas: Array[String] = data.split("_")
      datas(6) != "-1"
    }).map(
      data => {

        val datas: Array[String] = data.split("_")
        (datas(6), 1)
      }
    ).reduceByKey(_ + _)

    // 2、统计品类的下单数量。

    val orderCountData: RDD[(String, Int)] = rdd.filter(
      data => {
        val datas: Array[String] = data.split("_")
        datas(8) != "null"
      }
    ).flatMap(
      data => {
        val datas: Array[String] = data.split("_")
        val ids: Array[String] = datas(8).split(",")
      ids.map((_,1))
      }
    ).reduceByKey(_ + _)
    // 3、统计品类的支付数量
    val payCountData: RDD[(String, Int)] = rdd.filter(
      data => {
        val datas: Array[String] = data.split("_")
        datas(10) != "null"
      }
    ).flatMap(
      data => {
        val datas: Array[String] = data.split("_")
        //这里因为会有多个产品ID用逗号隔开，因此要将这一个字符串进行切割处理转换成字符串
        val ids: Array[String] = datas(10).split(",")
        ids.map((_,1))
      }
    ).reduceByKey(_ + _)

    // 现在有3个独立的统计结果，如果想让第一个统计结果影响其他的统计结果，比较麻烦
    // 先按照点击数量进行排序，如果相同，那么再按照下单排序，如果相同，再按照支付排序
    // TODO 元组的排序：先按照第一个数据进行排序，如果相同，按照第二个进行排序，如果相同，按照第三个进行排序
    // （品类，点击数量）
    // （品类，下单数量）
    // （品类，支付数量）
    // (品类，（点击数量，下单数量，支付数量 ）)

    val dataRDD: RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Int]))] =
      clickCountData.cogroup(orderCountData, payCountData)
    //dataRDD.collect().foreach(println)

    val mapDataRDD = dataRDD.mapValues {
      case ( clickIter, orderIter, payIter ) => {

        var clickCnt = 0
        var orderCnt = 0
        var payCnt = 0

        val iterator1: Iterator[Int] = clickIter.iterator
        if (iterator1.hasNext) {
          clickCnt = iterator1.next()
        }

        val iterator2: Iterator[Int] = orderIter.iterator
        if (iterator2.hasNext) {
          orderCnt = iterator2.next()
        }

        val iterator3: Iterator[Int] = payIter.iterator
        if (iterator3.hasNext) {
          payCnt = iterator3.next()
        }

        ( clickCnt, orderCnt, payCnt )
      }
    }
    //TODO 4、将统计的结果进行取前十名
    val top10 = mapDataRDD.sortBy(_._2, false).take(10)
    top10.foreach(println)

    //TODO 2、关闭Spark环境的连接
    sc.stop()

  }

}
